package com.tang.watermake;

import com.tang.partition.MyPartitioner;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 消息堆积到一个子任务导致下游窗口的水位线一直不更新
 *
 * @author tang
 * @since 2023/6/15 22:08
 */
public class WatermarkIdlenessDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(2);
        environment.socketTextStream("192.168.70.141", 7777)
                .partitionCustom(new MyPartitioner(), value -> value)
                .map(Integer::valueOf)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Integer>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Integer>() {
                                    @Override
                                    public long extractTimestamp(Integer element, long recordTimestamp) {
                                        return element * 1000L;
                                    }
                                }).withIdleness(Duration.ofSeconds(10L))
                )
                .keyBy(r -> r % 2)
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, ProcessWindowFunction<Integer, String, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {

                        out.collect(elements.toString());
                    }
                }).print();

        environment.execute();

    }

}
